package conlon.rocketmq.mq.listen;



import conlon.rocketmq.mq.ProxyModel;
import conlon.rocketmq.utils.BeansUtils;
import conlon.rocketmq.utils.ProtoBufSerializer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author conlon
 * @Description
 * @Date 2021/1/6 20:02
 **/
@Slf4j
public class MessageListen implements MessageListenerConcurrently {

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        try {
            MessageExt ext = list.get(0);
            ProxyModel proxyModel = ProtoBufSerializer.deSerialize(ext.getBody() , ProxyModel.class);
            log.info("MsgId : {} , proxyModel : {} " , ext.getMsgId() , proxyModel);
            Object object = BeansUtils.getBean(Class.forName(proxyModel.getClassName()));
            Method isMethod = ReflectionUtils.findMethod(object.getClass(), proxyModel.getMethodName(), proxyModel.getParameterTypes());
            assert isMethod != null;
            ReflectionUtils.invokeMethod(isMethod, object, proxyModel.getArgs());
        } catch (Exception e) {
            log.info("MessageListen-consumeMessage-Exception : ", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}

